package connect

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"encoding/json"
	"github.com/sirupsen/logrus"
	"go.uber.org/zap"
	"gochat/api/rpc"
	"gochat/config"
	"gochat/pkg/stickpackage"
	"gochat/proto"
	"net"
	"strings"
	"time"
)

const maxInt = 1<<31 - 1

func init() {
	rpc.InitLogicRpcClient()
}

func (c *Connect) InitTcpServer() error {

	aTcpAddr := strings.Split(config.Conf.Connect.ConnectTcp.Bind, ",")
	cpuNum := config.Conf.Connect.ConnectBucket.CpuNum
	var (
		addr     *net.TCPAddr
		listener *net.TCPListener
		err      error
	)
	for _, ipPort := range aTcpAddr {
		if addr, err = net.ResolveTCPAddr("tcp", ipPort); err != nil {
			zap.L().Error("server_tcp ResolveTCPAddr error:%s", zap.Error(err))
			return err
		}
		if listener, err = net.ListenTCP("tcp", addr); err != nil {
			zap.L().Error("net.ListenTCP(tcp, %s),error(%v)",
				zap.String("",ipPort),
				zap.Error(err))
			return err
		}
		zap.L().Info("start tcp listen at:%s", zap.String("",ipPort))
		// cpu core num
		for i := 0; i < cpuNum; i++ {
			go c.acceptTcp(listener)
		}
	}
	return nil
}

// tcp
func (c *Connect) acceptTcp(listener *net.TCPListener) {
	var (
		conn *net.TCPConn
		err  error
		r    int
	)
	connectTcpConfig := config.Conf.Connect.ConnectTcp
	for {
		if conn, err = listener.AcceptTCP(); err != nil {
			zap.L().Error("listener.Accept(\"%s\") error(%v)",
				zap.String("",listener.Addr().String()),
				zap.Error(err))

			return
		}
		// 设置keep alive，client==服务器ping包检查
		if err = conn.SetKeepAlive(connectTcpConfig.KeepAlive); err != nil {
			zap.L().Error("conn.SetKeepAlive() error:%s",zap.Error(err))
			return
		}
		// 设置接收buf
		if err := conn.SetReadBuffer(connectTcpConfig.ReceiveBuf); err != nil {
			zap.L().Error("conn.SetReadBuffer() error:%s",zap.Error(err))
			return
		}
		//set SendBuf
		if err := conn.SetWriteBuffer(connectTcpConfig.SendBuf); err != nil {
			zap.L().Error("conn.SetWriteBuffer() error:%s", zap.Error(err))
			return
		}
		go c.ServeTcp(DefaultServer, conn, r)
		if r++; r == maxInt {
			r = 0
		}
	}
}

func (c *Connect) ServeTcp(server *Server, conn *net.TCPConn, r int) {
	var ch *Channel
	ch = NewChannel(server.Options.BroadcastSize)
	ch.connTcp = conn
	go c.writeDataToTcp(server, ch)
	go c.readDataFromTcp(server, ch)
}

func (c *Connect) readDataFromTcp(s *Server, ch *Channel) {
	defer func() {
		zap.L().Info("start exec disConnect ...")
		if ch.Room == nil || ch.userId == 0 {

			logrus.Infof("roomId and userId eq 0")
			_ = ch.connTcp.Close()
			return
		}
		zap.L().Info("exec disConnect ...")
		disConnectRequest := new(proto.DisConnectRequest)
		disConnectRequest.RoomId = ch.Room.Id
		disConnectRequest.UserId = ch.userId
		s.Bucket(ch.userId).DeleteChannel(ch)
		if err := s.operator.DisConnect(disConnectRequest); err != nil {
			zap.L().Warn("DisConnect rpc err :%s", zap.Error(err))
		}
		if err := ch.connTcp.Close(); err != nil {
			zap.L().Warn("DisConnect close tcp conn err :%s", zap.Error(err))
		}
		return
	}()
	// scanner
	scannerPackage := bufio.NewScanner(ch.connTcp)
	scannerPackage.Split(func(data []byte, atEOF bool) (advance int, token []byte, err error) {
		if !atEOF && data[0] == 'v' {
			if len(data) > stickpackage.TcpHeaderLength {
				packSumLength := int16(0)
				_ = binary.Read(bytes.NewReader(data[stickpackage.LengthStartIndex:stickpackage.LengthStopIndex]), binary.BigEndian, &packSumLength)
				if int(packSumLength) <= len(data) {
					return int(packSumLength), data[:packSumLength], nil
				}
			}
		}
		return
	})
	scanTimes := 0
	for {
		scanTimes++
		if scanTimes > 3 {
			zap.L().Info("scannedPack times is:%d",zap.Int("",scanTimes))
			break
		}
		for scannerPackage.Scan() {
			scannedPack := new(stickpackage.StickPackage)
			err := scannedPack.Unpack(bytes.NewReader(scannerPackage.Bytes()))
			if err != nil {
				zap.L().Error("scan tcp package err:%s",zap.Error(err))
				break
			}
			//get a full package
			var connReq proto.ConnectRequest
			zap.L().Info("get a tcp message :%s",zap.Any("scannedPack",scannedPack))
			var rawTcpMsg proto.SendTcp
			if err := json.Unmarshal([]byte(scannedPack.Msg), &rawTcpMsg); err != nil {
				zap.L().Error("tcp message struct %+v",zap.Any("rawTcpMsg",rawTcpMsg))
				break
			}
			zap.L().Info("json unmarshal,raw tcp msg is:%+v",zap.Any("rawTcpMsg",rawTcpMsg))
			if rawTcpMsg.AuthToken == "" {
				zap.L().Error("tcp s.operator.Connect no authToken")
				return
			}
			if rawTcpMsg.RoomId <= 0 {
				zap.L().Error("tcp roomId not allow lgt 0")
				return
			}
			switch rawTcpMsg.Op {
			case config.OpBuildTcpConn:
				connReq.AuthToken = rawTcpMsg.AuthToken
				connReq.RoomId = rawTcpMsg.RoomId
				//fix
				connReq.ServerId = config.Conf.Connect.ConnectTcp.ServerId
				userId, err := s.operator.Connect(&connReq)

				zap.L().Info("tcp s.operator.Connect userId is :%d",zap.Int("userId",userId))
				if err != nil {
					zap.L().Error("tcp s.operator.Connect error %s",zap.Error(err))
					return
				}
				if userId == 0 {
					zap.L().Error("tcp Invalid AuthToken ,userId empty")
					return
				}
				b := s.Bucket(userId)
				//添加 一个 bucket
				err = b.Put(userId, connReq.RoomId, ch)
				if err != nil {
					zap.L().Error("tcp conn put room err: %s",zap.Error(err))
					_ = ch.connTcp.Close()
					return
				}
			case config.OpRoomSend:
				//发送消息到聊天室
				req := &proto.Send{
					Msg:          rawTcpMsg.Msg,
					FromUserId:   rawTcpMsg.FromUserId,
					FromUserName: rawTcpMsg.FromUserName,
					RoomId:       rawTcpMsg.RoomId,
					Op:           config.OpRoomSend,
				}
				code, msg := rpc.RpcLogicObj.PushRoom(req)
				zap.L().Info("tcp conn push msg to room,err code is:%d,err msg is:%s",
					zap.Int("code",code),
					zap.String("code",msg))
			}
		}
		if err := scannerPackage.Err(); err != nil {
			zap.L().Error("tcp get a err package:%s",zap.Error(err))
			return
		}
	}
}

func (c *Connect) writeDataToTcp(s *Server, ch *Channel) {
	//ping time default 54s
	ticker := time.NewTicker(DefaultServer.Options.PingPeriod)
	defer func() {
		ticker.Stop()
		_ = ch.connTcp.Close()
		return
	}()
	pack := stickpackage.StickPackage{
		Version: stickpackage.VersionContent,
	}
	for {
		select {
		case message, ok := <-ch.broadcast:
			if !ok {
				_ = ch.connTcp.Close()
				return
			}
			pack.Msg = message.Body
			pack.Length = pack.GetPackageLength()
			//发送消息
			zap.L().Info("send tcp msg to conn:%s",zap.String("",pack.String()))
			if err := pack.Pack(ch.connTcp); err != nil {
				zap.L().Error("connTcp.write message err:%s", zap.Error(err))
				return
			}
		case <-ticker.C:
			zap.L().Info("connTcp.ping message,send")
			//send a ping msg ,if error , return
			pack.Msg = []byte("ping msg")
			pack.Length = pack.GetPackageLength()
			if err := pack.Pack(ch.connTcp); err != nil {
				//send ping msg to tcp conn
				return
			}
		}
	}
}
